package com.chief.tableApi;


import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class TestTableFromDataSream {

    public static void main(String[] args) throws Exception {

//        EnvironmentSettings env = EnvironmentSettings.newInstance().inStreamingMode().useAnyPlanner().build();
//        TableEnvironment tableEnv = TableEnvironment.create(env);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStream<String> dataStream = env.fromElements("Alice", "Bob", "John");

        Table table = tableEnv.fromDataStream(dataStream);

        tableEnv.createTemporaryView("test", table);
        Table resultTable = tableEnv.sqlQuery("SELECT UPPER(f0) FROM test");

        // interpret the insert-only Table as a DataStream again
        DataStream<Row> resultStream = tableEnv.toDataStream(resultTable);

        // add a printing sink and execute in DataStream API
        resultStream.print();
        env.execute();

    }

}
